{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 数据处理Part 2\n",
    "* 处理用户的属性，通过用户的rat, 构造统计特征，max，min, avg, medium\n",
    "\n",
    "#### 需要每次构建模型都执行"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#!python --version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import pymysql\n",
    "import pymysql.cursors\n",
    "from functools import reduce\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import uuid\n",
    "import datetime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#获取一个数据库连接，注意如果是UTF-8类型的，需要制定数据库\n",
    "#port 必须是数字不能为字符串\n",
    "def get_connection():\n",
    "    return pymysql.connect(host='rm-2zeqqm6994abi7b6dqo.mysql.rds.aliyuncs.com',\n",
    "                               user='noone',\n",
    "                               password='Huawei12#$',\n",
    "                               db='recsys',\n",
    "                               port=3306,\n",
    "                               charset ='utf8',\n",
    "                               use_unicode=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def insert_or_update_one_userproex(userid, rmax, rmin, ravg, rmedium, rcount, rsum):\n",
    "    print('insert_or_update_one_userproex start')\n",
    "    is_insert = True\n",
    "    connection = get_connection()\n",
    "    try:\n",
    "        sql = 'select count(*) from userproex where userid=\\'%s\\'' % userid\n",
    "        with connection.cursor() as cursor:\n",
    "            cursor.execute(sql)\n",
    "        connection.commit()\n",
    "        res_cnt = cursor.fetchall()[0][0]\n",
    "        if res_cnt > 0:\n",
    "            is_insert = False\n",
    "    except Exception as e:\n",
    "        print('query for exist info error'+str(e))\n",
    "        \n",
    "    if is_insert:\n",
    "        sql = 'insert into userproex(userid, rmax, rmin, ravg, rcount, rsum, rmedian) values(\\'%s\\', %s, %s, %s, %s, %s, %s)' % (userid, rmax, rmin, ravg, rcount, rsum, rmedium)\n",
    "    else:\n",
    "        sql = 'update userproex set rmax=%s, rmin=%s, ravg=%s, rmedian=%s, rcount=%s, rsum=%s where userid=\\'%s\\'' % (rmax, rmin, ravg, rmedium, rcount, rsum, userid)\n",
    "    \n",
    "    try:\n",
    "        with connection.cursor() as cursor:\n",
    "            cursor.execute(sql)\n",
    "        connection.commit()\n",
    "        print(('insert userid:' if is_insert else 'update userid:') + str(userid) + ' success.')\n",
    "    except Exception as e:\n",
    "        print(e)\n",
    "        connection.close()\n",
    "    connection.close()\n",
    "    \n",
    "#insert_or_update_one_userproex('99', 9.1234567, 3.9, 5, 7, 9, 1.1111111)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def insert_or_update_one_userproex_4_median(userid, rmax, rmin, ravg, rmedian, rcount, rsum, connection):\n",
    "    print('insert_or_update_one_userproex_4_median start')\n",
    "    is_insert = True\n",
    "    try:\n",
    "        sql = 'select count(*) from userproex_new where userid=\\'%s\\'' % userid\n",
    "        with connection.cursor() as cursor:\n",
    "            cursor.execute(sql)\n",
    "        res_cnt = cursor.fetchall()[0][0]\n",
    "        if res_cnt > 0:\n",
    "            is_insert = False\n",
    "    except Exception as e:\n",
    "        print('query for exist info error'+str(e))\n",
    "        \n",
    "    if is_insert:\n",
    "        print('异常数据'+str(userid))\n",
    "        return\n",
    "    else:\n",
    "        sql = 'update userproex_new set rmedian=%s where userid=\\'%s\\'' % (rmedian, userid)\n",
    "    \n",
    "    try:\n",
    "        with connection.cursor() as cursor:\n",
    "            cursor.execute(sql)\n",
    "        print(('insert userid:' if is_insert else 'update userid:') + str(userid) + ' success.')\n",
    "    except Exception as e:\n",
    "        print(e)\n",
    "        connection.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "## 增量处理数据包含max,min..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def process_per_user(userid, insert_or_update_func):\n",
    "    connection = get_connection()\n",
    "    \n",
    "    rmedian = 0\n",
    "    rmax = 0\n",
    "    rmin = 0\n",
    "    ravg = 0\n",
    "    rcount = 0\n",
    "    rsum = 0\n",
    "    \n",
    "    sql = 'select rating from comment_new where userid=\\'%s\\'' % userid\n",
    "    try:\n",
    "        with connection.cursor() as cursor:\n",
    "            count = cursor.execute(sql)\n",
    "        rlist = cursor.fetchall()\n",
    "        rmedian = np.median(np.array(list(map(lambda x:x[0], rlist))).astype(float))\n",
    "    except Exception as e:\n",
    "        print(e)\n",
    "        connection.close()\n",
    "    \n",
    "    sql = 'select max(rating) as rmax, min(rating) as rmin, avg(rating) as ravg, count(rating) as rcount, sum(rating) as rsum from comment_new where userid=\\'%s\\'' % userid\n",
    "    try:\n",
    "        count = 0\n",
    "        with connection.cursor() as cursor:\n",
    "            count = cursor.execute(sql)\n",
    "        if count > 0:\n",
    "            r = cursor.fetchone()\n",
    "            rmax, rmin, ravg, rcount, rsum = r[0], r[1], r[2], r[3], r[4]\n",
    "    except Exception as e:\n",
    "        print(e)\n",
    "        connection.close()\n",
    "    insert_or_update_func(userid=userid, rmax=rmax, rmin=rmin, ravg=ravg, rmedium=rmedian, rcount=rcount, rsum=rsum)\n",
    "    connection.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "## 全量处理数据，只处理median，由于max,min等都有内置函数，所以直接sql处理"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def process_per_user_median(userid, insert_or_update_func, connection):\n",
    "    rmedian = 0\n",
    "    rmax = 0\n",
    "    rmin = 0\n",
    "    ravg = 0\n",
    "    rcount = 0\n",
    "    rsum = 0\n",
    "    sql = 'select rating from comment_new where userid=\\'%s\\'' % userid\n",
    "    try:\n",
    "        with connection.cursor() as cursor:\n",
    "            count = cursor.execute(sql)\n",
    "        rlist = cursor.fetchall()\n",
    "        rmedian = np.median(np.array(list(map(lambda x:x[0], rlist))).astype(float))\n",
    "    except Exception as e:\n",
    "        print(e)\n",
    "        connection.close()\n",
    "    \n",
    "    insert_or_update_func(userid=userid, rmax=rmax, rmin=rmin, ravg=ravg, rmedian=rmedian, rcount=rcount, rsum=rsum, connection=connection)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 全量处理数据\n",
    "\n",
    "* 1.执行sql\n",
    "\n",
    "    -- delete from userproex_new\n",
    "    -- INSERT INTO userproex_new (userid, rmax, rmin, ravg, rcount, rsum) (select userid, max(rating) as rmax, min(rating) as rmin, avg(rating) as ravg, count(rating) as rcount, sum(rating) as rsum from comment_new group by userid having userid is not null)\n",
    "* 2.执行中位数更新"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def get_user_list_in_comment(is_all_or_sample, process_func, insert_or_update_func):\n",
    "    connection = get_connection()\n",
    "    if is_all_or_sample:\n",
    "        sql = 'select DISTINCT userid from comment_new where userid is not null and userid in (select userid from userproex_new where rmedian=0)'\n",
    "    else:\n",
    "        sql = 'select DISTINCT userid from comment_new where userid is not null and userid in (select userid from userproex_new where rmedian=0) limit 0,10'\n",
    "        \n",
    "    try:\n",
    "        with connection.cursor() as cursor:\n",
    "            cursor.execute(sql)\n",
    "        connection.commit()\n",
    "        process_size = 0\n",
    "        while True:\n",
    "            r = cursor.fetchone()\n",
    "            if r == None:\n",
    "                break\n",
    "            process_func(r[0], insert_or_update_func, connection)\n",
    "            process_size = process_size + 1\n",
    "            if process_size >= 1000:\n",
    "                connection.commit()\n",
    "                process_size = 0\n",
    "        connection.commit()\n",
    "    except Exception as e:\n",
    "        print(e)\n",
    "        connection.close()\n",
    "    connection.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2018-12-19 11:10:13.615361\n"
     ]
    }
   ],
   "source": [
    "print(datetime.datetime.now())\n",
    "start_time = datetime.datetime.now()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2018-12-19 11:00:42.943586\n",
      "update userid:cf2349f9c01f9a5cd4050aebd30ab74f success.\n",
      "update userid:6abe55491bf68a3873106a1945ad631e success.\n",
      "update userid:4092d171faf4e2bcdfb9c6fc230fc594 success.\n",
      "update userid:0d295b2e85b5a47dd7c0d74176153598 success.\n",
      "update userid:536a5ea5ea7a5a4259a7c2c2082301ab success.\n",
      "update userid:3262d48df5d75e3452f0f16b313b7808 success.\n",
      "update userid:28383bc85352627c623c4dbc7e9d3b0a success.\n",
      "update userid:6cfe37b82c14779526a2bb86a560aaa4 success.\n",
      "update userid:3d24df53ce1b3e1b06333e8f8a6913ef success.\n",
      "update userid:272c489a90cc7949961e6da4c634e377 success.\n",
      "2018-12-19 11:00:54.798482\n",
      "0:00:11.854899\n"
     ]
    }
   ],
   "source": [
    "get_user_list_in_comment(False, process_per_user_median, insert_or_update_one_userproex_4_median)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2018-12-19 11:10:31.568195\n",
      "0:00:17.952835\n"
     ]
    }
   ],
   "source": [
    "print(datetime.datetime.now())\n",
    "end_time = datetime.datetime.now()\n",
    "print(end_time - start_time)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## 增量处理数据\n",
    "\n",
    "* 根据支持，传入user list\n",
    "\n",
    "* 根据user list, 更新max，min, avg, sum, count， median等值"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def get_user_list_in_log(is_all_or_sample, process_func, insert_or_update_func):\n",
    "    connection = get_connection()\n",
    "    if is_all_or_sample:\n",
    "        sql = 'select DISTINCT userid from comment_new_log where userid is not null and active=1'\n",
    "    else:\n",
    "        sql = 'select DISTINCT userid from comment_new_log where userid is not null and active=1 limit 0,10'\n",
    "        \n",
    "    try:\n",
    "        with connection.cursor() as cursor:\n",
    "            cursor.execute(sql)\n",
    "        connection.commit()\n",
    "        while True:\n",
    "            r = cursor.fetchone()\n",
    "            if r == None:\n",
    "                break\n",
    "            process_func(r[0], insert_or_update_func)\n",
    "            with connection.cursor() as cursor:\n",
    "                update_sql = 'update comment_new_log set active=0 where userid=\\'%s\\'' % r[0]\n",
    "                cursor.execute(update_sql)\n",
    "            connection.commit()\n",
    "    except Exception as e:\n",
    "        print(e)\n",
    "        connection.close()\n",
    "    connection.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#get_user_list_in_log(False, process_per_user, insert_or_update_one_userproex)"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [default]",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
